package sprs;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLType;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.regex.Pattern;

import org.apache.commons.collections.IteratorUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.rosuda.REngine.REXP;
import org.rosuda.REngine.Rserve.RConnection;

import scala.Tuple2;

public final class Trend {
	private static final Pattern SPACE = Pattern.compile("\\|");

	private static Logger log = Logger.getLogger(Trend.class);

	private static class TupleComparator implements Comparator<Tuple2<Date, Double>>, Serializable {
		// @Override
		public int compare(Tuple2<Date, Double> tuple1, Tuple2<Date, Double> tuple2) {
			return tuple1._1.compareTo(tuple2._1);
		}
	}

	public static void main(String[] args) throws Exception {

		SparkConf sparkConf = new SparkConf().setAppName("Trend");
		String currentFile = "2014-11-29.csv";
		if (args.length < 1) {
			args = new String[] { "/data/rmr" };
			sparkConf.setMaster("local[4]");
			// System.err.println("Usage: JavaWordCount <file>");
			// System.exit(1);
		}

		JavaSparkContext ctx = new JavaSparkContext(sparkConf);

		List<String> curUsidList = new ArrayList();
		try {
			JavaRDD<String> currentUsids = null;
			currentUsids = ctx.textFile(args[0] + "/FIFI_USID_METRIC_TREND_20141114.csv", 1);
			currentUsids = currentUsids.map(new Function<String, String>() {
				// @Override
				public String call(String s) {

					String a[] = SPACE.split(s);
					return a[11];
				}

			});

			currentUsids = currentUsids.distinct();
			curUsidList = currentUsids.collect();

			log.info("Total current Usids =" + curUsidList.size());

		} catch (Exception e) {

		}

		JavaRDD<String> lines = ctx.textFile(args[0]);
		log.info("Total lines =" + lines.count());

		// JavaRDD<String> lines = ctx.wholeTextFiles(args[0], 1);

		// JavaRDD<String> lines =
		// ctx.textFile("/data/rmr/01-01-2012.csv").union(ctx.textFile("/data/rmr/02-01-2012.csv"));
		final List<String> curUsidListFinal = curUsidList;

		JavaRDD<String> headerRemoved = lines.filter(new Function<String, Boolean>() {
			// @Override
			public Boolean call(String s) {

				String a[] = SPACE.split(s);
				if (curUsidListFinal.size() != 0 && !curUsidListFinal.contains(a[11]))
					return false;
				if (s.contains("MET"))
					return false;
				if (s.contains("LTE"))
					return false;
				if (s.contains("DROP"))
					return true;
				else
					return false;
			}

		});
		
		/*
		 * JavaRDD<String> quoteRemoved = headerRemoved .map(new
		 * Function<String, String>() { // @Override public String call(String
		 * s) { return s.replace("\"", ""); } });
		 */
		log.info("Lines (DROP)=" + headerRemoved.count());

		/*
		 * JavaPairRDD<String, Double> ones = headerRemoved.mapToPair(new
		 * PairFunction<String, String, Double>() { // @Override public
		 * Tuple2<String, Double> call(String s) {
		 * 
		 * String a[] = SPACE.split(s); // System.out.println(a[0] + "..." + s);
		 * double d = -1; if (!a[10].isEmpty()) d = Double.parseDouble(a[10]);
		 * 
		 * // Date date = new //
		 * java.text.SimpleDateFormat("dd-MMM-yyyy").parse(a[6]); // Tuple2 key
		 * = new Tuple2<String, Date>(a[0], date); return new Tuple2<String,
		 * Double>(a[0] + a[6], d);
		 * 
		 * } });
		 */

		JavaPairRDD<Tuple2<String, String>, Tuple2<Date, Double>> keyValue = headerRemoved
				.mapToPair(new PairFunction<String, Tuple2<String, String>, Tuple2<Date, Double>>() {

					public Tuple2<Tuple2<String, String>, Tuple2<Date, Double>> call(String s) throws Exception {
						try {
							String a[] = SPACE.split(s);
							double d = -1;
							if (!a[10].isEmpty())
								d = Double.parseDouble(a[10]);
							Date date = new java.text.SimpleDateFormat("dd-MMM-yyyy").parse(a[6]);
							Tuple2<String, String> key = new Tuple2(a[0], a[11]); // 11
																					// Usid.
																					// 0
																					// Metric
							Tuple2<Date, Double> value = new Tuple2(date, d);

							return new Tuple2<Tuple2<String, String>, Tuple2<Date, Double>>(key, value);
						} catch (Exception e) {
							throw new Exception("line=" + s);
						}

					}

				});
		log.info("key pair=" + keyValue.count());

		JavaPairRDD<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>> keyValues = keyValue.groupByKey();
		log.info("key pairs =" + keyValues.count());
		processwithR(keyValues);
		// ctx.stop();
	}

	public static void processwithR(JavaPairRDD<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>> keyValues) {

		// keyValues.fl
		JavaPairRDD<Tuple2<String, String>, List> keyValues2 = keyValues
				.mapToPair(new PairFunction<Tuple2<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>>, Tuple2<String, String>, List>() {

					public Tuple2<Tuple2<String, String>, List> call(Tuple2<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>> t)
							throws Exception {
						Tuple2<String, String> key = t._1;
						Iterator<Tuple2<Date, Double>> it = t._2.iterator();
						List<Tuple2<Date, Double>> list2 = IteratorUtils.toList(it);
						List<Tuple2<Date, Double>> decomOutput = new ArrayList<Tuple2<Date, Double>>();
						Collections.sort(list2, new TupleComparator());
						double prev = list2.get(0)._2; // dealing with missing
														// data.
						/*for (Tuple2<Date, Double> el : list2) {
							log.debug(el._1 + "..." + el._2);
							 
						}*/
						
						for (Tuple2<Date, Double> el : list2) {
							log.debug(el._1);
							if (el._2 != -1) {
								prev = el._2;
								break;
							}
						}
						List<Double> forR = new ArrayList<Double>();
						for (Tuple2<Date, Double> el : list2) {
							if (el._2 != -1) {
								forR.add(el._2);
								prev = el._2;
							} else {
								forR.add(prev);
							}
							// System.out.println(el._1 + "  " + el._2);
							// output.add( new Tuple2<Date, Double>(el._1,
							// el._2*2));
						}
						double[] inputToR = ArrayUtils.toPrimitive(forR.toArray(new Double[0]));

						// if (true) return new Tuple2<Tuple2<String, String>,
						// Iterable<Tuple2<Date, Double>>>(key, decomOutput);

						double[] outputFromR = null;
						RConnection rc = null;
						try {
							rc = RConnectionPool.getPool().borrowObject();

							rc.assign("d", inputToR);
							rc.voidEval("ts <- ts(d, start=c(2011, 1, 1), freq=365.25)");
							rc.voidEval("my.tbats <- tbats(ts, seasonal.periods=c(7, 365.25), use.trend=TRUE, use.parallel=TRUE)");
							rc.voidEval("components = tbats.components(my.tbats)");

							REXP rexp = null;

							rexp = rc.eval("resid(my.tbats)+components[,2]+components[,3]");
							outputFromR = rexp.asDoubles();

							log.debug("");
							log.debug(ArrayUtils.toString(outputFromR));
							log.debug(key._1 + " " + key._2);
							// System.out.println(ArrayUtils.toString(outputD));

						} catch (Exception e) {
							log.error("catched error", e);
							return new Tuple2<Tuple2<String, String>, List>(key, new ArrayList());

						} finally {

							if (rc != null)
								RConnectionPool.getPool().returnObject(rc);
						}

						ArrayList retList = new ArrayList();
						try {
							/*
							 * for (int i = 0; i < list2.size(); i++) {
							 * Tuple2<Date, Double> temp = list2.get(i); if
							 * (temp._2 < 0) decomOutput.add(temp); else
							 * decomOutput.add(new Tuple2<Date, Double>(temp._1,
							 * outputFromR[i])); }
							 */
							// Current decomposed value.

							if (outputFromR[outputFromR.length - 1] != -1) {
								retList.add(list2.get(list2.size() - 1)._1); // Date...
								retList.add(new Double(outputFromR[outputFromR.length - 1]));

							} else {
								retList.add(null);
								retList.add(null);
							}

							double shortTermOrg[] = Arrays.copyOfRange(inputToR, outputFromR.length - 17, outputFromR.length);
							double shortTerm[] = Arrays.copyOfRange(outputFromR, outputFromR.length - 17, outputFromR.length);
							double shortTermForPattern[] = Arrays.stream(shortTerm).filter(x -> x != -1.0).toArray();

							Tuple2<Double, double[]> shortTermCoff = Util.coff(shortTermForPattern);

							retList.add(Arrays.toString(shortTermForPattern));
							retList.add(shortTermCoff);

							// retList.add(Arrays.toString(shortTermForPattern));

							// System.out.println("Short term coff=" + coff._1);

							if (log.isDebugEnabled()) {
								// log.debug("");
								log.debug("Short term coff=" + shortTermCoff._1);
								log.debug(Arrays.toString(shortTermCoff._2));
								log.debug(Arrays.toString(shortTermOrg));
								log.debug(Arrays.toString(shortTermForPattern));
							}

							double longTermOrg[] = Arrays.copyOfRange(inputToR, outputFromR.length - 60, outputFromR.length);
							double longTerm[] = Arrays.copyOfRange(outputFromR, outputFromR.length - 60, outputFromR.length);
							double longTermForPattern[] = Arrays.stream(longTerm).filter(x -> x != -1.0).toArray();

							Tuple2<Double, double[]> longTermCoff = Util.coff(longTermForPattern);

							retList.add(Arrays.toString(longTermForPattern));
							retList.add(longTermCoff);

							if (log.isDebugEnabled()) {
								log.debug("Long term coff=" + longTermCoff._1);
								log.debug(Arrays.toString(longTermCoff._2));
								log.debug(Arrays.toString(longTermOrg));
								log.debug(Arrays.toString(longTermForPattern));

							}
						} catch (Exception e) {
							log.error(e);
						}

						return new Tuple2<Tuple2<String, String>, List>(key, retList);
					}

				});

		log.info("Got result keyvalues=" + keyValues2.count());
		insertToDB(keyValues2.collect());
		
		// keyValues.saveAsObjectFile("objectfile.obj");
		if (log.isDebugEnabled() && false) {
			List<Tuple2<Tuple2<String, String>, List>> list = keyValues2.collect();
			for (Tuple2<Tuple2<String, String>, List> t : list) {
				System.out.println(t._1._1 + ".." + t._1._2);
				Iterator<Tuple2<Date, Double>> it = t._2.iterator();
				List<Tuple2<Date, Double>> list2 = IteratorUtils.toList(it);

				Collections.sort(list2, new TupleComparator());
				for (Tuple2<Date, Double> el : list2) {
					System.out.println(el._1 + "  " + el._2);
				}

				/*
				 * while (it.hasNext()) { Tuple2<Date, Double> el = it.next();
				 * System.out.println(el._1 + "  " + el._2); }
				 */
			}
		}

	}

	final static String INSERT_SQL = "MERGE INTO NRS.FIFI_USID_METRIC_TREND_DECOMP n "
			+ "USING  (SELECT  ? metricname, ? usid, ? theDate, ? METRICPCT_LAST, ? METRICLIST_ST,  ? PMSTEP_CORREL_ST, ? PMSTEP_PATTERN_ST,  "
			+ "? METRICLIST_LT, ? PMSTEP_CORREL_LT,  ? PMSTEP_PATTERN_LT from DUAL) s "
			+ "ON (n.usid = s.usid and n.metricname = s.metricname and n.periodend = theDate )  "
			+ "WHEN MATCHED THEN update set n.METRICPCT_LAST=s.METRICPCT_LAST, "
			+ "n.METRICLIST_ST = s.METRICLIST_ST, n.PMSTEP_CORREL_st =s.PMSTEP_CORREL_st, n.PMSTEP_PATTERN_St = s.PMSTEP_PATTERN_ST, "
			+ "n.METRICLIST_lT = s.METRICLIST_lT, n.PMSTEP_CORREL_lt =s.PMSTEP_CORREL_lt, n.PMSTEP_PATTERN_lT = s.PMSTEP_PATTERN_lt "
			+ "  WHEN NOT MATCHED THEN  "
			+ "INSERT (usid, metricname, periodend, METRICPCT_LAST, PMSTEP_CORREL_ST,  METRICLIST_ST,  PMSTEP_PATTERN_ST,  "
			+ "  PMSTEP_CORREL_LT, METRICLIST_LT,  PMSTEP_PATTERN_LT)"
			+ " VAlUES(s.usid, s.metricname, s.thedate, s.METRICPCT_LAST, s.PMSTEP_CORREL_ST,  s.METRICLIST_ST,  s.PMSTEP_PATTERN_ST,  "
			+ "  s.PMSTEP_CORREL_LT, s.METRICLIST_LT,  s.PMSTEP_PATTERN_LT )";

	public static void insertToDB(List<Tuple2<Tuple2<String, String>, List>> ret) {

		Connection con = null;
		PreparedStatement ps = null;
		int count = 0;
		try {
			con = Connectors.getFifiCon();
			con.setAutoCommit(false);
			ps = con.prepareStatement(INSERT_SQL);
			for (Tuple2<Tuple2<String, String>, List> row : ret) {

				List dataList = row._2;
				/*for (int i=0; i<dataList.size(); i++) {
					Object obj = dataList.get(i);
					System.out.println(i+"..."+obj.getClass() + "" + obj);
				}*/
				if (dataList.get(0) == null || dataList.get(1) == null || (Double) dataList.get(1) == -1)
					continue;
				int k = 1;
				//System.out.println(row._1._1 +"..."+row._1._2);
				
				ps.setString(k++, row._1._1);
				ps.setString(k++, row._1._2);

				java.sql.Date d = new java.sql.Date(((java.util.Date) dataList.get(0)).getTime());

				ps.setDate(k++, d);
				ps.setDouble(k++, (Double) dataList.get(1));
				
				ps.setString(k++, dataList.get(2) + "");
				Tuple2<Double, double[]> t = (Tuple2<Double, double[]>) dataList.get(3);
				ps.setDouble(k++, t._1);
				ps.setString(k++, Arrays.toString(t._2));

				ps.setString(k++, dataList.get(4) + "");
				Tuple2<Double, double[]> t2 = (Tuple2<Double, double[]>) dataList.get(5);
				ps.setDouble(k++, t2._1);
				ps.setString(k++, Arrays.toString(t2._2));
				ps.addBatch();
				count ++;
				if (count%3000==0){
					ps.executeBatch();
					con.commit();
				}

			}
			ps.executeBatch();
			con.commit();

		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			

		}
	}

	public static void processwithR0(JavaPairRDD<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>> keyValues) {

		// keyValues.fl
		JavaPairRDD<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>> keyValues2 = keyValues
				.mapToPair(new PairFunction<Tuple2<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>>, Tuple2<String, String>, Iterable<Tuple2<Date, Double>>>() {

					public Tuple2<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>> call(
							Tuple2<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>> t) throws Exception {
						Tuple2<String, String> key = t._1;
						Iterator<Tuple2<Date, Double>> it = t._2.iterator();
						List<Tuple2<Date, Double>> list2 = IteratorUtils.toList(it);
						List<Tuple2<Date, Double>> decomOutput = new ArrayList<Tuple2<Date, Double>>();
						Collections.sort(list2, new TupleComparator());
						double prev = list2.get(0)._2; // dealing with missing
														// data.
						for (Tuple2<Date, Double> el : list2) {
							if (el._2 != -1) {
								prev = el._2;
								break;
							}
						}
						List<Double> forR = new ArrayList<Double>();
						for (Tuple2<Date, Double> el : list2) {
							if (el._2 != -1) {
								forR.add(el._2);
								prev = el._2;
							} else {
								forR.add(prev);
							}
							// System.out.println(el._1 + "  " + el._2);
							// output.add( new Tuple2<Date, Double>(el._1,
							// el._2*2));
						}
						double[] inputToR = ArrayUtils.toPrimitive(forR.toArray(new Double[0]));

						// if (true) return new Tuple2<Tuple2<String, String>,
						// Iterable<Tuple2<Date, Double>>>(key, decomOutput);

						double[] outputFromR = null;
						RConnection rc = null;
						try {
							rc = RConnectionPool.getPool().borrowObject();
							// rc = new RConnection("ulpd326.madc.att.com");

							// log.info("Got into R?");
							rc.assign("d", inputToR);
							rc.voidEval("ts <- ts(d, start=c(2011, 1, 1), freq=365.25)");
							rc.voidEval("my.tbats <- tbats(ts, seasonal.periods=c(7, 365.25), use.trend=TRUE, use.parallel=TRUE)");
							rc.voidEval("components = tbats.components(my.tbats)");

							REXP rexp = null;
							// REXP rexp = rc.eval("components[,1]");
							// double[] original = rexp.asDoubles();
							// outputFromR = original;
							rexp = rc.eval("resid(my.tbats)+components[,2]+components[,3]");
							outputFromR = rexp.asDoubles();
							/*
							 * double[] resid = rexp.asDoubles(); outputFromR =
							 * new double[resid.length]; rexp =
							 * rc.eval("components[,2]"); double[] level =
							 * rexp.asDoubles(); rexp =
							 * rc.eval("components[,3]"); double[] scope =
							 * rexp.asDoubles();
							 * 
							 * outputFromR = new double[level.length];
							 * 
							 * for (int i = 0; i < outputFromR.length; i++) {
							 * outputFromR[i] = resid[i] + level[i] + scope[i];
							 * }
							 */

							log.debug("");
							log.debug(ArrayUtils.toString(outputFromR));
							log.debug(key._1 + " " + key._2);
							// System.out.println(ArrayUtils.toString(outputD));

						} catch (Exception e) {
							log.error("catched error", e);
							return new Tuple2<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>>(key, decomOutput);

						} finally {

							if (rc != null)
								RConnectionPool.getPool().returnObject(rc);
						}

						try {
							for (int i = 0; i < list2.size(); i++) {
								Tuple2<Date, Double> temp = list2.get(i);
								if (temp._2 < 0)
									decomOutput.add(temp);
								else
									decomOutput.add(new Tuple2<Date, Double>(temp._1, outputFromR[i]));
							}
							double shortTermOrg[] = Arrays.copyOfRange(inputToR, outputFromR.length - 17, outputFromR.length);
							double shortTerm[] = Arrays.copyOfRange(outputFromR, outputFromR.length - 17, outputFromR.length);
							double shortTermPattern[] = Arrays.stream(shortTerm).filter(x -> x != -1.0).toArray();

							Tuple2<Double, double[]> coff = Util.coff(shortTerm);
							// System.out.println("Short term coff=" + coff._1);

							if (log.isDebugEnabled()) {
								// log.debug("");
								log.debug("Short term coff=" + coff._1);
								log.debug(Arrays.toString(coff._2));
								log.debug(Arrays.toString(shortTermOrg));
								log.debug(Arrays.toString(shortTermPattern));
							}

							double longTermOrg[] = Arrays.copyOfRange(inputToR, outputFromR.length - 60, outputFromR.length);
							double longTerm[] = Arrays.copyOfRange(outputFromR, outputFromR.length - 60, outputFromR.length);
							double longTermPattern[] = Arrays.stream(longTerm).filter(x -> x != -1.0).toArray();

							coff = Util.coff(longTerm);
							if (log.isDebugEnabled()) {
								log.debug("Long term coff=" + coff._1);
								log.debug(Arrays.toString(coff._2));
								log.debug(Arrays.toString(longTermOrg));
								log.debug(Arrays.toString(longTermPattern));

							}
						} catch (Exception e) {
							log.error(e);
						}
						return new Tuple2<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>>(key, decomOutput);
					}

				});

		log.info("Got result keyvalues=" + keyValues2.count());
		// keyValues.saveAsObjectFile("objectfile.obj");
		if (log.isDebugEnabled() && false) {
			List<Tuple2<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>>> list = keyValues2.collect();
			for (Tuple2<Tuple2<String, String>, Iterable<Tuple2<Date, Double>>> t : list) {
				System.out.println(t._1._1 + ".." + t._1._2);
				Iterator<Tuple2<Date, Double>> it = t._2.iterator();
				List<Tuple2<Date, Double>> list2 = IteratorUtils.toList(it);

				Collections.sort(list2, new TupleComparator());
				for (Tuple2<Date, Double> el : list2) {
					System.out.println(el._1 + "  " + el._2);
				}

				/*
				 * while (it.hasNext()) { Tuple2<Date, Double> el = it.next();
				 * System.out.println(el._1 + "  " + el._2); }
				 */
			}
		}

	}

}